package org.hard.netty.c4;

import lombok.extern.slf4j.Slf4j;
import org.hard.netty.ByteBufferUtil;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author: 王林博
 * @date: 2022年08月13日 18:21
 */
@Slf4j
public class MultiThreadServer {
    public static void main(String[] args) throws IOException {
        Thread.currentThread().setName("boss");
        ServerSocketChannel socketChannel = ServerSocketChannel.open();
        socketChannel.configureBlocking(false);
        Selector boss = Selector.open();
        final SelectionKey register = socketChannel.register(boss, 0, null);
        register.interestOps(SelectionKey.OP_ACCEPT);
        socketChannel.bind(new InetSocketAddress(8080));
        // 创建固定数量的worker
        Worker[] worker = new Worker[2];
        for (int i = 0; i < worker.length; i++) {
            worker[i] = new Worker("worker-" + i);
        }
        AtomicInteger index = new AtomicInteger();

        while (true){
            boss.select();
            final Iterator<SelectionKey> iterator = boss.selectedKeys().iterator();
            while (iterator.hasNext()){
                final SelectionKey key = iterator.next();
                iterator.remove();
                if(key.isAcceptable()){
                    final SocketChannel accept = socketChannel.accept();
                    accept.configureBlocking(false);
                    // 关联 selector
                    log.debug("connected...{}",accept.getRemoteAddress());
                    log.debug("before register...{}",accept.getRemoteAddress());
                    worker[index.getAndIncrement() % worker.length].register(accept);
//                    worker.register(accept);
                    log.debug("after register...{}",accept.getRemoteAddress());
                }else if(key.isReadable()){

                }

            }
        }

    }
   static class Worker implements Runnable{
        private Thread thread;
        private Selector selector;
        private String name;
        private volatile  boolean start = false;

        private ConcurrentLinkedQueue<Runnable> queue = new ConcurrentLinkedQueue<>();

        public Worker(String name) {
            this.name = name;
        }
        // 初始化线程
        public void register(SocketChannel accept) throws IOException {
            if(!start){
                thread = new Thread(this,name);
                selector = Selector.open();
                thread.start();
                start = true;
            }

            // 向队列添加任务，但是这个任务并没有执行
//            queue.add(()->{
//                try {
//                    accept.register(selector,SelectionKey.OP_READ,null);
//                } catch (ClosedChannelException e) {
//                    e.printStackTrace();
//                }
//            });
            // 唤醒 selector
            selector.wakeup();
            accept.register(selector,SelectionKey.OP_READ,null);
        }

        @Override
        public void run() {
            while (true){
                try {
                    selector.select();// 阻塞状态  使用 wakeup可以唤醒 selector
//                    final Runnable task = queue.poll();
//                    if( task != null){
//                        task.run(); // 执行 accept.register(selector,SelectionKey.OP_READ,null);
//                    }
                    final Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                    while (iterator.hasNext()){
                        final SelectionKey key = iterator.next();
                        iterator.remove();
                        if(key.isReadable()){
                            ByteBuffer buffer = ByteBuffer.allocate(16);
                            ByteBufferUtil.debugAll(buffer);
                            final SocketChannel socketChannel = (SocketChannel) key.channel();
                            socketChannel.read(buffer);
                            buffer.flip();
                            ByteBufferUtil.debugAll(buffer);
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
